package com.guanzi.big_data_mq_process.listener;

import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.read.listener.ReadListener;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.extension.service.IService;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Function;
import java.util.function.Predicate;

/**
 * 通用的模板excel导入监听器
 *
 * @author 管子
 * @date 2025/6/14 13:57
 * @description: EasyExcelListener 类
 */
@Slf4j
public class EasyExcelListener<PO, BO> implements ReadListener<BO> {

    // 线程内自己的实体类数据
    private final ThreadLocal<List<PO>> entityList = ThreadLocal.withInitial(ArrayList::new);

    // 线程内自己的批量插入任务集合
    private final ThreadLocal<List<CompletableFuture<Boolean>>> dbFutureList = ThreadLocal.withInitial(ArrayList::new);

    private final int batchSize;

    private final IService<PO> poService;

    // BO转为PO的逻辑
    private final Function<BO, PO> convertFunction;

    private final Predicate<BO> boPredicate;

    private final ThreadPoolExecutor threadPoolExecutor;

    public EasyExcelListener(Predicate<BO> boPredicate, Function<BO, PO> convertFunction, IService<PO> poService, int batchSize, ThreadPoolExecutor threadPoolExecutor) {
        this.boPredicate = boPredicate;
        this.convertFunction = convertFunction;
        this.poService = poService;
        this.batchSize = batchSize;
        this.threadPoolExecutor = threadPoolExecutor;
    }

    // 每解析一行Excel数据，EasyExcel会转成指定的BO对象并回调该方法
    @Override
    public void invoke(BO data, AnalysisContext context) {
        if (!boPredicate.test(data)) {
            throw new RuntimeException("接收到的BO导入对象校验异常!");
        }
        PO po = convertFunction.apply(data);
        entityList.get().add(po);
        if (entityList.get().size() >= batchSize) {
            asyncSaveBatchData();
        }
    }

    @Override
    public void doAfterAllAnalysed(AnalysisContext context) {
        log.info("线程: {} 处理完了一个Sheet数据", Thread.currentThread().getName());
        if (!entityList.get().isEmpty()) {
            asyncSaveBatchData();
        }
    }

    /**
     * 异步批量保存的逻辑
     */
    private void asyncSaveBatchData() {
        // 注意entityList是调用 asyncSaveBatchData() 的线程的ThreadLocal里的数据，不能直接交给线程池使用
        if (!entityList.get().isEmpty()) {
            // 使用JSON转换深拷贝
            List<PO> copyEntityList = JSON.parseArray(JSON.toJSONString(entityList.get()), poService.getEntityClass());

            CompletableFuture<Boolean> dbFuture = CompletableFuture.supplyAsync(() -> {
                log.info("线程: {} 插入 {} 条数据", Thread.currentThread().getName(), copyEntityList.size());
                return poService.saveBatch(copyEntityList, copyEntityList.size());
            }, threadPoolExecutor).exceptionally((e) -> {
                log.error("线程: {} 批量插入数据异常: {}", Thread.currentThread().getName(), e.getMessage());
                return false;
            });

            dbFutureList.get().add(dbFuture);
            entityList.get().clear();
        }
    }

    /**
     * 获取所有的批量插入任务，同时清除ThreadLocal
     * 该方法为一次性方法
     */
    public List<CompletableFuture<Boolean>> getDbFutureList() {
        // 注意! 这里不能深拷贝，CompleteFuture没有实现序列化接口，会导致拷贝后的join方法调用卡死! 切记！！！
        List<CompletableFuture<Boolean>> newCompleteFutureList = new ArrayList<>(dbFutureList.get());
        dbFutureList.get().clear();
        return newCompleteFutureList;
    }

}
